package com.crazymaker.springcloud.message.service.impl;

import com.crazymaker.push.platform.contract.dto.MsgDTO;
import com.crazymaker.springcloud.common.constants.SessionConstants;
import com.crazymaker.springcloud.common.util.JsonUtil;
import com.crazymaker.springcloud.common.util.ThreadUtil;
import com.crazymaker.springcloud.distribute.idGenerator.impl.IdWorker;
import com.crazymaker.springcloud.standard.context.SpringContextUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.admin.TopicStatsTable;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.apache.rocketmq.tools.admin.DefaultMQAdminExt;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.List;
import java.util.UUID;

@Component
@Slf4j
public class MessageRelayService implements ApplicationContextAware {

    private String brokerClusterName = "DefaultCluster";


    @Value("${rocketmq.nameServer}")
    private String rocketmqAddress;
    @Value("${rocketmq.consumer.group}")
    private String consumerGroup;
    @Value("${rocketmq.producer.group}")
    private String producerGroup;

    @Value("${rocketmq.producer.topic}")
    private String topic;


    @Value("${rocketmq.producer.relayTopic}")
    private String relayTopic;

    @Value("${rocketmq.consumer.relayGroup}")
    private String consumerRelayGroup;


    @Value("${rocketmq.producer.concurrent}")
    private int concurrent = 1;

    @Value("${rocketmq.producer.retryTimes}")
    private int retryTimes = 1;

    @Value("${rocketmq.producer.maxWaited}")
    private int maxWaited = 100000;

    private volatile boolean inited = false;


    DefaultMQPushConsumer consumer = null;


    @Resource
    MessagePushServiceImpl mqMessageService;

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        SpringContextUtil.setContext(applicationContext);
        initRelayConsumerAndProducer();
    }

    private synchronized void initRelayConsumerAndProducer() {
        if (inited) return;
        inited = true;

        startConsumer();
        startProducer();
    }

    DefaultMQProducer producer = null;

    private void startProducer() {
        producer = new DefaultMQProducer(producerGroup);
        //指定NameServer地址
        producer.setNamesrvAddr(rocketmqAddress); //修改为自己的
        producer.setRetryTimesWhenSendFailed(retryTimes);
        String instanceName = producerGroup + "-" + IdWorker.getInstance().getId();
        producer.setInstanceName(instanceName);
        producer.setProducerGroup(producerGroup);
        /**
         * Producer对象在使用之前必须要调用start初始化，初始化一次即可
         * 注意：切记不可以在每次发送消息时，都调用start方法
         */
        try {
            producer.start();
            log.info("product {} start ...", instanceName);
        } catch (MQClientException e) {
            e.printStackTrace();
        }
    }


    private void startConsumer() {


        /**
         * Consumer Group,非常重要的概念，后续会慢慢补充
         */
        consumer = new DefaultMQPushConsumer(consumerRelayGroup);
        consumer.setMessageModel(MessageModel.CLUSTERING);

        //指定NameServer地址，多个地址以 ; 隔开
        consumer.setNamesrvAddr(rocketmqAddress); //修改为自己的
//        consumer.setNamesrvAddr("192.168.116.115:9876;192.168.116.116:9876"); //修改为自己的


/**
 * 设置Consumer第一次启动是从队列头部开始消费还是队列尾部开始消费
 * 如果非第一次启动，那么按照上次消费的位置继续消费
 */
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);

        try {
            consumer.subscribe(relayTopic, "*");
        } catch (MQClientException e) {
            e.printStackTrace();
            return;
        }

        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {

                try {
                    for (int i = 0; i < msgs.size(); i++) {
                        MessageExt msg = msgs.get(i);
//                    String content = new String(msg.getBody());
//                    log.info("收到消息：{}", msg.getMsgId() + " " + msg.getTopic() + " " + msg.getTags() + " " + content);

                        try {


                        MsgDTO dto = JsonUtil.jsonBytes2Object(msg.getBody(), MsgDTO.class);
                        dto.setMsgId(Long.valueOf(msg.getKeys()));
                        System.out.println("dto = " + dto);

                        String appId = msg.getUserProperty(SessionConstants.USER_IDENTIFIER);
                        dto.setAppId(Long.valueOf(appId));
                        mqMessageService.saveAndDispatchMessage(dto);
                        }catch (Exception e)
                        {
//                            e.printStackTrace();
                        }
                    }
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;

                } catch (Throwable throwable) {
                    throwable.printStackTrace();
                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;

                }

            }
        });
        try {
            consumer.start();
        } catch (MQClientException e) {
            e.printStackTrace();
        }

    }


    public void createTopic(String topic) {

        boolean createResult = false;
        DefaultMQAdminExt mqAdminExt = new DefaultMQAdminExt();
        mqAdminExt.setInstanceName(UUID.randomUUID().toString());
        mqAdminExt.setNamesrvAddr(rocketmqAddress);
        try {
            mqAdminExt.start();
            mqAdminExt.createTopic(brokerClusterName, topic, 4);
        } catch (Exception e) {
        }

        long startTime = System.currentTimeMillis();
        while (!createResult) {
            createResult = checkTopicExist(mqAdminExt, topic);
            if (System.currentTimeMillis() - startTime < 10 * 1000) {
                ThreadUtil.sleepMilliSeconds(100);
            } else {
                log.error(String.format("timeout,but create topic[%s] failed!", topic));
                break;
            }
        }

        mqAdminExt.shutdown();
    }

    private static boolean checkTopicExist(DefaultMQAdminExt mqAdminExt, String topic) {
        boolean createResult = false;
        try {
            TopicStatsTable topicInfo = mqAdminExt.examineTopicStats(topic);
            createResult = !topicInfo.getOffsetTable().isEmpty();
        } catch (Exception e) {
        }

        return createResult;
    }


    public SendResult send(Message msg) {
        try {
//            msg.setTopic(topic);

            return producer.send(msg);
        } catch (Exception e) {
            e.printStackTrace();
        }
        return null;
    }
}
